#include "sys.h"
#include "log.h"
#include "dial.h"
#include "cJSON.h"
#include "kafka.h"
#include "ping_tactics.h"

void* do_ping_tactics(void* arg)
{
    ping_tactics_t* ping = (ping_tactics_t*)arg;
    pingTactics_t* ptr;
    struct pingTactics_head* head = &ping->head;
    cJSON *result, *obj, *arr;

    dial_arg_t dial_arg = { 0 };
    dial_result_t* dial_result = NULL;
    ping_result_t* ping_result = NULL;
    tcpping_result_t* tcpping_result = NULL;
    char tmp_str[32] = { 0 };
    char topic[32] = { 0 };
    char *post_data = NULL, reportTime[32], collectTime[32];
    if (ping->count == 0)
        return NULL;
    result = cJSON_CreateObject();
    cJSON_AddStringToObject(result, "dialTaskId", ping->taskId);
    cJSON_AddStringToObject(result, "version", ping->version);
    cJSON_AddStringToObject(result, "indexType", "ping");
    get_timestring(reportTime);
    cJSON_AddStringToObject(result, "reportTime", reportTime);
    get_timestring(collectTime);
    cJSON_AddStringToObject(result, "collectTime", collectTime);
    // pthread_mutex_lock(&ping->mutex);
    arr = cJSON_CreateArray();
    TAILQ_FOREACH(ptr, head, entry)
    {

        obj = cJSON_CreateObject();
        if (!strcmp(ptr->type, "tcp")) {
            dial_arg.in.tcpping_arg.ip = wd_ipaddr(ptr->host);
            dial_arg.in.tcpping_arg.port = htons(ptr->port);
            dial_arg.in.tcpping_arg.counts = (ptr->times);
            dial_arg.in.tcpping_arg.timeout = (ptr->timeout);
            dial_result = (dial_result_t*)dial_tcpping(&dial_arg);
            tcpping_result_print(&dial_result->out.tcpping_result);

            tcpping_result = &dial_result->out.tcpping_result;
            sprintf(tmp_str, "%.2f", tcpping_result->lost_percent);
            cJSON_AddStringToObject(obj, "loss", tmp_str);
            sprintf(tmp_str, "%.2f", tcpping_result->jitter);
            cJSON_AddStringToObject(obj, "jitter", tmp_str);
            sprintf(tmp_str, "%.2f", tcpping_result->latency);
            cJSON_AddStringToObject(obj, "avgDelayTime", tmp_str);
        } else {
            dial_arg.in.ping_arg.ip = wd_ipaddr(ptr->host);
            dial_arg.in.ping_arg.blocksize = (ptr->size);
            dial_arg.in.ping_arg.counts = (ptr->times);
            dial_arg.in.ping_arg.timeout = (ptr->timeout);
            dial_result = (dial_result_t*)dial_ping(&dial_arg);
            ping_result_print(&dial_result->out.ping_result);

            ping_result = &dial_result->out.ping_result;
            sprintf(tmp_str, "%.2f", ping_result->lost_percent);
            cJSON_AddStringToObject(obj, "loss", tmp_str);
            sprintf(tmp_str, "%.2f", ping_result->jitter);
            cJSON_AddStringToObject(obj, "jitter", tmp_str);
            sprintf(tmp_str, "%.2f", ping_result->latency);
            cJSON_AddStringToObject(obj, "avgDelayTime", tmp_str);
        }
        strcpy(topic, ptr->kafkaTopic);
        cJSON_AddStringToObject(obj, "host", ptr->host);
        cJSON_AddStringToObject(obj, "businessTag", ptr->businessTag);
        cJSON_AddItemToArray(arr, obj);
        free(dial_result);
    }
    cJSON_AddItemToObject(result, "pingList", arr);
    post_data = cJSON_Print(result);
    log_debug("PING post_data:%s", post_data);
    kafka_send(post_data, strlen(post_data), ping->kafka_url, topic);
    free(post_data);
    cJSON_Delete(result);
    // pthread_cond_signal(&ping->cond);
    // pthread_mutex_unlock(&ping->mutex);
    return NULL;
}